"""
Case Type   : HTAP行列融合特性
Case Name   : HTAP行列融合特性--不支持防篡改模式表，合理报错
Create At   : 2025/11/5
Owner       : sungang14
Description :
    1.建表
    2.插入数据
    3.全表转换
    4.主备查询
    5.取消转换
    6.部分列转换
    7.主备查询
    8.取消转换
    9.清理环境
Expect      :
    1.执行成功
    2.执行成功
    3.合理报错
    4.主机查询成功，备机查询为空
    5.合理报错
    6.合理报错
    7.主机查询成功，备机查询为空
    8.执行成功
    9.执行成功
History     :
"""

import os
import unittest
import time

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant
from yat.test import macro


class HTAPIntegerTypes(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.sta1_sh = CommonSH('Standby1DbUser')
        self.sta2_sh = CommonSH('Standby2DbUser')
        self.constant = Constant()

        self.pg_hba_backup = None
        text = '----备份pg_hba.conf文件;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'cp {macro.DB_INSTANCE_PATH}/pg_hba.conf {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)

        text = '----配置主机信任连接;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'gs_guc reload -N all -I all -h "host all all {self.pri_sh.node.db_host}/32 trust";' \
              f'gs_guc reload -N all -I all -h "host all all {self.sta1_sh.node.db_host}/32 trust";' \
              f'gs_guc reload -N all -I all -h "host all all {self.sta2_sh.node.db_host}/32 trust"'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)
        self.table_name_1 = 't_htap_0024_1'
        self.table_name_2 = 't_htap_0024_2'
        self.table_name_3 = 't_htap_0024_3'
        self.table_name_4 = 't_htap_0024_4'
        self.schema_name = 's_htap_0024'

    def test_integer_types_htap(self):
        text = '----step1:建表;expect:执行成功----'
        self.log.info(text)
        sql_cmd = f"drop schema if exists {self.schema_name} cascade; create schema {self.schema_name} with blockchain;"
        q_res = self.pri_sh.execut_db_sql(sql_cmd)
        self.log.info(q_res)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, q_res, f'执行失败:{text}')
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, q_res, f'执行失败:{text}')

        sql_cmd = f'drop table if exists {self.table_name_1} cascade;drop table if exists {self.table_name_2} cascade;drop table if exists {self.table_name_3} cascade; drop table if exists {self.table_name_4} cascade;'
        res = self.pri_sh.execut_db_sql(sql_cmd)
        self.assertEqual(res.count(self.constant.TABLE_DROP_SUCCESS), 4, f'执行失败:{text}')
        sql = f"""
                create table {self.schema_name}.{self.table_name_1}(
                    c1 int,
                    c2 char(30)
                );
                create table {self.schema_name}.{self.table_name_2}(
                    c1 int,
                    c2 char(30)
                )partition by range (c1)
                (
                    partition t_htap_0024_2_p1 values less than (50),
                    partition t_htap_0024_2_p2 values less than (1000)
                );
                create table {self.schema_name}.{self.table_name_3}(
                    c1 int,
                    c2 char(30)
                )with (orientation = column);
                create table {self.schema_name}.{self.table_name_4}(
                    c1 int,
                    c2 char(30)
                )with (orientation = column)
                partition by range (c1)
                (
                    partition t_htap_0024_4_p1 values less than (50),
                    partition t_htap_0024_4_p2 values less than (1000)
                );"""
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(q_res.count(self.constant.TABLE_CREATE_SUCCESS), 4, f'执行失败:{text}')

        text = '----step2:插入数据;expect:执行成功----'
        self.log.info(text)
        sql = f"insert into {self.schema_name}.{self.table_name_1} values (generate_series(1,100), generate_series(1,100)||'hello world');" \
              f"insert into {self.schema_name}.{self.table_name_2} values (generate_series(1,100), generate_series(1,100)||'hello world');" \
              f"insert into {self.schema_name}.{self.table_name_3} values (generate_series(1,100), generate_series(1,100)||'hello world');" \
              f"insert into {self.schema_name}.{self.table_name_4} values (generate_series(1,100), generate_series(1,100)||'hello world');" \
              f"select pg_sleep(5);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertEqual(q_res.count("INSERT 0 100"), 4, f'执行失败:{text}')

        text = '----step3:全表转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_1} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assert1_msg = f"Table type not support for HTAP"
        self.assert2_msg = f"column-store relation doesn't support this ALTER yet"
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_2} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_3} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

        text = '----step4:主备查询;expect:查询成功----'
        self.log.info(text)
        sql_cmd1 = f"select count(*) from {self.schema_name}.{self.table_name_1};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_2};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_3};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_4};"
        q_res = self.pri_sh.execut_db_sql(sql_cmd1)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')
        sql_cmd2 = f"set enable_imcsscan to on;" \
                   f"select count(*) from {self.schema_name}.{self.table_name_1};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_2};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_3};" \
                   f"select count(*) from {self.schema_name}.{self.table_name_4};"
        q_res = self.sta1_sh.execut_db_sql(sql_cmd2)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')

        text = '----step5:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_1} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assert3_msg = f"rel not populated, no need to be unpopulate"
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_2} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_3} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

        text = '----step6:部分列转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_1} imcstored(c1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_2} imcstored(c1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_3} imcstored(c1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} imcstored(c1);"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

        text = '----step7:主备查询;expect:查询成功----'
        self.log.info(text)
        q_res = self.pri_sh.execut_db_sql(sql_cmd1)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')
        q_res = self.sta1_sh.execut_db_sql(sql_cmd2)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')

        text = '----step8:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_1} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_2} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert3_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_3} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

        text = '----step9:指定分区转换;expect:执行合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_2} modify partition t_htap_0024_2_p1 imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert1_msg, q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} modify partition t_htap_0024_4_p1 imcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

        text = '----step10:主备查询;expect:查询成功----'
        self.log.info(text)
        q_res = self.pri_sh.execut_db_sql(sql_cmd1)
        self.log.info(f'主节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')
        q_res = self.sta1_sh.execut_db_sql(sql_cmd2)
        self.log.info(f'备节点查询结果:\n{q_res}')
        self.assertEqual(q_res.count("100"), 4, f'执行失败:{text}')

        text = '----step11:取消转换;expect:合理报错----'
        self.log.info(text)
        sql = f"alter table {self.schema_name}.{self.table_name_2} modify partition t_htap_0024_2_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(f"not populated", q_res, f'执行失败:{text}')
        sql = f"alter table {self.schema_name}.{self.table_name_4} modify partition t_htap_0024_4_p1 unimcstored;"
        q_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(q_res)
        self.assertIn(self.assert2_msg, q_res, f'执行失败:{text}')

    def tearDown(self):
        text = '----step12:清理环境;expect:执行成功----'
        self.log.info(text)
        sql_cmd = f"drop schema if exists {self.schema_name} cascade;"
        q_res = self.pri_sh.execut_db_sql(sql_cmd)
        self.log.info(q_res)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, q_res, f'执行失败:{text}')

        sql_cmd = f'drop table if exists {self.table_name_1} cascade;drop table if exists {self.table_name_2} cascade;drop table if exists {self.table_name_3} cascade; drop table if exists {self.table_name_4} cascade;'
        res = self.pri_sh.execut_db_sql(sql_cmd)
        self.assertEqual(res.count(self.constant.TABLE_DROP_SUCCESS), 4, f'执行失败:{text}')

        text = '----恢复pg_hba.conf文件;expect:执行成功----'
        self.log.info(text)
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'cp {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test {macro.DB_INSTANCE_PATH}/pg_hba.conf;' \
              f'gs_om -t restart'
        self.log.info(cmd)
        exec_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(exec_res)
        time.sleep(30)
        cmd = f'source {macro.DB_ENV_PATH};gs_om -t status --detail'
        status_res = self.pri_sh.node.sh(cmd).result()
        self.log.info(f'数据库集群状态:\n{status_res}')
        cmd = f'source {macro.DB_ENV_PATH};rm -f {macro.DB_INSTANCE_PATH}/pg_hba.conf.backup.test'
        self.pri_sh.node.sh(cmd).result()
        self.log.info('pg_hba.conf备份文件已清理')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
